package interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class TypeInterceptor implements Interceptor {

    //声明一个存放事件的集合
    private List<Event> addHeaderEvent;

    @Override
    public void initialize() {
        //初始化一个存放事件的集合
        addHeaderEvent = new ArrayList<>();

    }

    //单个事件拦截
    @Override
    public Event intercept(Event event) {
        //获取事件的头信息
        Map<String, String> headers = event.getHeaders();
        //获取事件的body信息
        String body = new String(event.getBody());
        //根据事件的body信息中是否有”hello“来决定添加怎样的头信息
        if (body.contains("hello")) {
            headers.put("type", "hashello");
        } else {
            headers.put("type", "others");
        }
        return event;
    }

    //批量事件拦截
    @Override
    public List<Event> intercept(List<Event> list) {
        //清空events集合
        addHeaderEvent.clear();
        //遍历list
        for (Event event : list) {
            //给集合中的每个数据添加头信息
            addHeaderEvent.add(intercept(event));
        }
        return addHeaderEvent;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new TypeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
